package com.lahuan.config.kafka;

import com.alibaba.fastjson.JSON;

import com.lahuan.demo.entity.Order;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.*;
@Component
public class ConsumerImpl {
    private Properties properties;
    private KafkaConsumer<String, String> consumer;
    @Value("${spring.kafka.template.default-topic}")
    private  String topic  ;
    private Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>();
    private Set<String> orderSet = new HashSet<>();
    private volatile boolean flag = true;

    public ConsumerImpl() {
        properties = new Properties();
//        properties.put("enable.auto.commit", false);
//        properties.put("isolation.level", "read_committed");
//        properties.put("auto.offset.reset", "latest");
        //只用来重放
        properties.put("group.id", "replay-group");
        properties.put("bootstrap.servers", "localhost:9092");
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        consumer = new KafkaConsumer<>(properties);
    }


    public List<Order> consumeOrder(long offset,long len ) {
        ArrayList<Order> arrayList = new ArrayList<>();
        consumer.subscribe(Collections.singletonList(topic));
        //获取consumer所分配的分区信息
        Set<TopicPartition> assignments = consumer.assignment();
        if(assignments!=null && assignments.size()>0){
            Iterator<TopicPartition> iterator = assignments.iterator();
            TopicPartition partition = iterator.next();
            consumer.seek(partition,offset);
            ConsumerRecords<String, String> poll = consumer.poll(Duration.ofSeconds(60));
            long count = 0;
            for (ConsumerRecord<String,String> record : poll) {
                Order order = JSON.parseObject(record.value(), Order.class);
                arrayList.add(order);
                if(count >=len){
                    break;
                }
            }
        }
        return arrayList;
    }

    public void close() {
        if (this.flag) {
            this.flag = false;
        }
        consumer.close();
    }

    private void deduplicationOrder(Order order) {
        orderSet.add(order.getId().toString());
    }
}
